package cloud.xlh.digital_power_pay.config.rocketmq;

import cloud.xlh.infrastructure.mq.operate.BaseMQOperateParamIn;
import cloud.xlh.infrastructure.mq.operate.BaseMQOperateParamOut;
import cloud.xlh.infrastructure.mq.operate.MQOperateService;
import cloud.xlh.infrastructure.mq.operate.MQSendMessageFunctionInterface;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.nio.charset.StandardCharsets;
import java.util.Map;

/**
 * Created with IntelliJ IDEA.
 *
 * @ project name : xlh_osms
 * @ Author: XuLeHuang
 * @ Date: 2023/10/12 10:42
 * @ Description:
 */
@Service(value = "rocketMQOperateService")
public class RocketMQOperateService implements MQOperateService {
    @Resource
    private RocketMQTemplate rocketMQTemplate;

    @Override
    public BaseMQOperateParamOut sendMessagePostHandle(BaseMQOperateParamOut baseMQOperateParamOut) {
        RocketMQOperateParamOut paramOut = (RocketMQOperateParamOut) baseMQOperateParamOut;
        RocketMQOperateParamIn paramIn = (RocketMQOperateParamIn) paramOut.getPreHandleResult();
        Map diyData = (Map) paramIn.getDiyData();
        diyData.put("status", "000000");
        return paramOut;
    }


    @Override
    public BaseMQOperateParamOut sendMessagePreHandle(BaseMQOperateParamIn baseMQOperateParamIn, BaseMQOperateParamOut baseMQOperateParamOut) {
        RocketMQOperateParamIn paramIn = (RocketMQOperateParamIn) baseMQOperateParamIn;
        Map diyData = (Map) paramIn.getDiyData();
        String topic = (String) diyData.get("topic");
        String routingKey = (String) diyData.get("routingKey");
        String message = (String) diyData.get("message");
        paramIn.setTopic(topic);
        paramIn.setRoutingKey(routingKey);
        Message reallyMessage = new Message();
        reallyMessage.setBody(message.getBytes(StandardCharsets.UTF_8));
        paramIn.setMessage(reallyMessage);
        RocketMQOperateParamOut paramOut = (RocketMQOperateParamOut) baseMQOperateParamOut;
        paramOut.setPreHandleResult(paramIn);
        return paramOut;
    }

    public BaseMQOperateParamIn buildMQOperateParamIn(Map<String, String> paramInMap) {
        BaseMQOperateParamIn<Map<String, String>> paramIn = new RocketMQOperateParamIn();
        paramIn.setMqSendMessageFunctionInterface(new MQSendMessageFunctionInterface<RocketMQOperateParamOut, RocketMQOperateParamIn>() {
            @Override
            public RocketMQOperateParamOut sendMessage(RocketMQOperateParamIn in, RocketMQOperateParamOut out) {
                rocketMQTemplate.asyncSend(in.getTopic(),
                        in.getMessage(), new SendCallback() {
                            @Override
                            public void onSuccess(SendResult sendResult) {

                            }

                            @Override
                            public void onException(Throwable throwable) {

                            }
                        });
                return null;
            }
        });
        paramIn.setDiyData(paramInMap);
        return paramIn;
    }

    public void process(Map<String, String> paramInMap) {
        BaseMQOperateParamIn paramIn = buildMQOperateParamIn(paramInMap);
        BaseMQOperateParamOut<Map, Map, Map> paramOut = new RocketMQOperateParamOut();
        paramOut = (RocketMQOperateParamOut) sendMessage(paramIn, paramOut);
    }
}
